//
// Created by 33145 on 2023/6/2.
//

#include "tcp_connection.h"
#include "../common/rocket.h"
#include "../net/fdevent_group.h"
#include "../coder/tinypb_coder.h"
#include "../rpc/rpc_dispatcher.h"

namespace rocket {


  TCPConnection::TCPConnection(EventLoop::ptr event_loop, int fd, int buffer_size,
                               NetAddr::ptr peer_addr, NetAddr::ptr local_addr, TcpConnectionType connection_type /* = ByServer*/)
      : _event_loop(event_loop), _peer_addr(peer_addr), _local_addr(local_addr),
        _state(NotConnected), _fd(fd) {
    _in_buffer = std::make_shared<TCPBuffer>(buffer_size);
    _out_buffer = std::make_shared<TCPBuffer>(buffer_size);

    _connection_type = connection_type;

    _fd_event = FdEventGroup::GetFdEventGroup()->getFdEvent(fd);

    _fd_event->setNonBlock();
    // 客户端不需要主动监听read
    if (connection_type != TcpConnectionByClient) {
      _fd_event->listen(FdEvent::IN_EVENT, [this]() { onRead(); });
      _event_loop->addEpollEvent(_fd_event.get());
    }

    _coder = std::make_shared<TinyPBCoder>();
  }

  TCPConnection::~TCPConnection() {
    LOG_DEBUG("a connection ~ : [%s]", _peer_addr->toString().c_str());
  }

  void TCPConnection::onRead() {
    // 从Socket读字节到io_buffer
    if (_state != Connected) {
      LOG_ERROR("onRead error, client has already disconneced, addr[%s], clientfd[%d]", _peer_addr->toString().c_str(), _fd_event->getFd());
      return;
    }
    bool is_read_all = false;
    bool is_close = false;
    while(true) {
      if (_in_buffer->writeAble() == 0) {
        _in_buffer->resizeBuffer(2 * _in_buffer->getBuffer().size());
      }

      int read_count = _in_buffer->writeAble();
      int write_index = _in_buffer->writeIndex();

      auto rt = read(_fd, &(_in_buffer->getBuffer()[write_index]), read_count);
      LOG_DEBUG("success onRead %d bytes from addr[%s], client fd[%d]", rt, _peer_addr->toString().c_str(), _fd);
      if (rt > 0) {
        _in_buffer->moveWriteIndex((int)rt);
        if (rt == read_count) {
          continue;
        } else if (rt < read_count) {
          is_read_all = true;
          break;
        }
      } else if (rt == 0) {
        is_close = true;
        break;
      } else if (rt == -1 && errno == EAGAIN) {
        is_read_all = true;
        break;
      }
    }
    if (is_close) {
      //TODO:
      LOG_INFO("peer closed, peer addr [%s], clientfd [%d]", _peer_addr->toString().c_str(), _fd);
      clear();
      return;
    }

    if (!is_read_all) {
      LOG_ERROR("not read all data");
    }

    // TODO: 简单的 echo, 后面补充 RPC 协议解析
    execute();
  }

  void TCPConnection::onWrite() {
    // echo 发送数据到client
    if (_state != Connected) {
      LOG_ERROR("onWrite error, client has already disconneced, addr[%s], clientfd[%d]", _peer_addr->toString().c_str(), _fd_event->getFd());
      return;
    }

    if (_connection_type == TcpConnectionByClient) {
      // 编码message得到字节流 将字节流写入buffer, 全部发送
      std::vector<AbstractProtocol::ptr> messages;
      for (auto [abs_ptr, _] : _write_dones) {
        messages.push_back(abs_ptr);
      }
      _coder->encode(messages, _out_buffer);
    }

    bool is_write_all = false;
    while(true) {
      if (_out_buffer->readAble() == 0) {
        LOG_DEBUG("no data need to send to client [%s]", _peer_addr->toString().c_str());
        is_write_all = true;
        break;
      }
      int write_size = _out_buffer->readAble();
      int read_index = _out_buffer->readIndex();
      int rt = write(_fd, &(_out_buffer->getBuffer()[read_index]), write_size);

      if (rt >= write_size) {
        LOG_DEBUG("no data need to send to client [%s]", _peer_addr->toString().c_str());
        is_write_all = true;
        _out_buffer->moveReadIndex(rt); // TODO: 这里感觉没处理发送到一半的情况啊，好像是默认要么全发要么发
        break;
      } else if (rt == -1 && errno == EAGAIN) {
        // 发送缓冲区已满，不能再发送了。
        // 这种情况我们等下次 fd 可写的时候再次发送数据即可
        LOG_ERROR("write data error, errno==EAGIN and rt == -1");
        break;
      }
      _out_buffer->moveReadIndex(rt);  // 如果一次没发完就调整一下索引
    }
    if (is_write_all) {
      _fd_event->cancel(FdEvent::OUT_EVENT);  // 为啥要取消write回调。。每次都要重新注册？
                                            // 因为如果去除就会一直触发可写事件唤醒epoll_wait
      _event_loop->addEpollEvent(_fd_event.get());
    }

    if (_connection_type == TcpConnectionByClient) {  // 客户端msg发送后执行相应回调
      for (const auto &[abs_ptr, func] : _write_dones) {
        func(abs_ptr);
      }
      _write_dones.clear();
    }
  }

  void TCPConnection::execute() {
    // 将 RPC 请求执行业务逻辑，获取 RPC 响应, 再把 RPC 响应发送回去
    // echo
    if (_connection_type == TcpConnectionByServer) {
      std::vector<AbstractProtocol::ptr> result;
      std::vector<AbstractProtocol::ptr> reply_message;
      _coder->decode(result, _in_buffer);  // 将消息解码为TinyPBProtocol集合
      for (size_t i = 0;  i < result.size(); ++i) {
        // 1. 针对每一个请求，调用 rpc 方法，获取响应 message
        LOG_INFO("success get request[%s] from client[%s]", result[i]->_msg_id.c_str(), _peer_addr->toString().c_str());
        std::shared_ptr<TinyPBProtocol> message = std::make_shared<TinyPBProtocol>();
        // message->m_pb_data = "hello. this is rocket rpc test data";
        // message->m_msg_id = result[i]->m_msg_id;

        RPCDispatcher::GetRPCDispatcher()->dispatch(result[i], message, this);
        reply_message.emplace_back(message);
      }
      // 2. 将响应 message 解码，放入到发送缓冲区，监听可写事件回包
      _coder->encode(reply_message, _out_buffer);

      _fd_event->listen(FdEvent::OUT_EVENT, [this]() { onWrite(); });
      _event_loop->addEpollEvent(_fd_event.get());

    } else {  // 客户端：解析消息中的msg_id， 执行回调
      std::vector<std::shared_ptr<AbstractProtocol>> result;
      _coder->decode(result, _in_buffer);

      for (const auto& msg_ptr : result) {
        std::string req_id = msg_ptr->getReqId();
        if (_read_dones.find(req_id) != _read_dones.end()) {
          _read_dones[req_id](msg_ptr);
        }
      }
      _fd_event->cancel(FdEvent::IN_EVENT);  // read事件执行完后需要取消可读事件监听
      _event_loop->addEpollEvent(_fd_event.get());
    }

  }

  void TCPConnection::clear() {
    // 处理一些关闭连接后的清理动作
    if (_state == Closed) {
      return;
    }
    _fd_event->cancel(FdEvent::IN_EVENT);
    _fd_event->cancel(FdEvent::OUT_EVENT);

    _event_loop->delEpollEvent(_fd_event.get());

    _state = Closed;
  }

  void TCPConnection::shutDown() {
    if (_state == Closed || _state == NotConnected) {
      return;
    }
    _state = HalfClosing; // 正在关闭（半关闭）
    ::shutdown(_fd, SHUT_RDWR);  // 调用shutdown后，意味着服务器不会再对该fd进行读写操作
                                 // 但客户端仍对该socket进行读写，服务端发送FIN报文，触发四次挥手第一阶段
                                 // 随后对端收到报文，也回复一个FIN报文
  }

  void TCPConnection::listenWrite() {
    _fd_event->listen(FdEvent::OUT_EVENT, [this](){ onWrite(); });
    _event_loop->addEpollEvent(_fd_event.get());
  }

  void TCPConnection::listenRead() {
    _fd_event->listen(FdEvent::IN_EVENT, [this](){ onRead(); });
    _event_loop->addEpollEvent(_fd_event.get());
  }

  void TCPConnection::pushSendMessage(const AbstractProtocol::ptr& message, const std::function<void(AbstractProtocol::ptr)>& done) {
    _write_dones.emplace_back(message, done);
  }

  void
  TCPConnection::pushReadMessage(const std::string &req_id, const std::function<void(AbstractProtocol::ptr)> &done) {
    _read_dones.insert(std::make_pair(req_id, done));
  }

}